package extract.document.general;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;
import common.config.DbTableNames;
import common.config.IndexTypeNames;
import common.dao.config.ServerConfigurer;
import common.dao.couchdb.CouchDBDOCFactory;
import common.dao.elastic.ElasticSearchDao;
import common.dao.mysql.MySqlDao;
import common.helper.json.JsonToJavaObject;
import extract.helper.CouchdbChangeHelper;
import extract.helper.DocUploadHelper;
import extract.helper.ElasticChangeHelper;
import extract.helper.TransactionHelper;
import org.ektorp.CouchDbConnector;
import org.ektorp.changes.DocumentChange;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.Map;

/**
 * Created by Common on 11/26/16.
 */
public class BaseIncreMode {

    public static final Logger logger = LoggerFactory.getLogger(BaseIncreMode.class);

    private final CouchdbChangeHelper couchdb_helper;
    private final CouchDBDOCFactory couch_factory;
    private final CouchDbConnector video_con;
    private final ElasticChangeHelper es_helper;
    private final DocUploadHelper docUploadHelper;
    private final ObjectMapper mapper;

    private final String es_index_name = IndexTypeNames.MediaDB.IndexName;

    public BaseIncreMode(CouchDBDOCFactory couch_factory,CouchdbChangeHelper couchdb_helper,
                         ElasticChangeHelper es_helper, DocUploadHelper docUploadHelper){
        this.couchdb_helper = couchdb_helper;
        this.couch_factory = couch_factory;
        this.video_con = this.couch_factory.getVideoCon();
        this.es_helper = es_helper;
        this.docUploadHelper = docUploadHelper;
        this.mapper = new ObjectMapper();
    }



    public boolean hasNext(){
        return this.couchdb_helper.isHaveChange();
    }

    public boolean updateToElastic(JsonNode doc){
        //this.es_dao.save_bulk(this.es_index_name, typeName, doc);
        return this.es_helper.saveToElastic(doc);
    }

    public boolean deleteFromElastic(String id){
        return this.es_helper.deleteFromElastic(id);
    }


    public JsonNode extractParentToChild(JsonNode child, JsonNode parent){

        return child;
    }

    public JsonNode updateNodeWithUploadInfo(JsonNode node, String couch_id, String couch_type){
        Map<String,String> info = this.docUploadHelper.getDocInfoByCouchId(couch_id, couch_type);
        ObjectNode indexNode = null;
        if(node.has("Indexdata")){
            indexNode = (ObjectNode)node.get("Indexdata");
        }
        else{
            indexNode = this.mapper.createObjectNode();
            ((ObjectNode)node).put("Indexdata", (ObjectNode)indexNode);
        }
        String copyright = info.get("copyright");
        String resource_type = info.get("resource_type");
        String video_format = info.get("video_format");
        indexNode.put("Copyright", copyright);
        indexNode.put("ResourceType", resource_type);
        indexNode.put("VideoFormat", video_format);
        return node;
    }


    public Map<String,String> getUploadInfoFromMysql(String couch_id, String couch_type){
        Map<String,String> info = this.docUploadHelper.getDocInfoByCouchId(couch_id, couch_type);
        return info;
    }


    public void insertOrModify(DocumentChange change, String id, String rev, int seq){
        //new_id = id;
        //new_serial = seq;
        String doc_json = change.getDoc();
        logger.info("id = {}, seq = {}\n", id, seq);
        //logger.info("doc_json = \n{}" ,doc_json);
        try{
            // current_doc
            JsonNode doc_node = JsonToJavaObject.loadJsonNode(doc_json);
            /*
            parent doc
             */
            if(doc_node.has("Metadata") == false){
                logger.error("[] not have Metadata!");
                // TODO: should store reversion
                this.couchdb_helper.insertNewestSeqNumber(id,seq,-1);
                return;
            }
            //boolean success = false;
            if(doc_node.get("Metadata").has("ParentID") == false){
                boolean success = updateToElastic(doc_node);
                int status = 1;
                if(success)
                    logger.info("save_new_doc_to_es success!");
                else {
                    status = -1;
                    logger.error("save_new_doc_to_es failed!");
                }
                this.couchdb_helper.insertNewestSeqNumber(id,seq, status );
            }
            else{
                String parent_id = doc_node.get("Metadata").get("ParentID").asText();
                logger.info("parent_id = {}", parent_id);

                // TODO: 目前暂时忽略 Parent
                //JsonNode parent_doc = this.video_con.get(JsonNode.class,parent_id);

                //String parent_str = JsonToJavaObject.jsonNodeToString(parent_doc);
                //logger.info("parent_json  = \n{}", parent_str);

                //String parent_rev = parent_doc.get("_rev").asText();
                //logger.info("parent_rev  = {}", parent_rev);

                /*
                extract, update to es
                 */
                //JsonNode new_node = extractParentToChild(doc_node, parent_doc);


                // TODO: 暂时不使用与parent合并后结果
                String parent_rev = rev;
                JsonNode merge_node = doc_node;

                // 使用upload数据更新
                JsonNode update_node = merge_node;
                try {
                    update_node = updateNodeWithUploadInfo(merge_node, id, "ignore_now");
                }
                catch (Exception e){
                    logger.error("ERROR! MSG = {}", e.toString());
                }


                boolean success = updateToElastic(update_node);

                int status = 1;
                if(success)
                    logger.info("save_new_doc_to_es success!");
                else {
                    status = -1;
                    logger.error("save_new_doc_to_es failed!");
                }
                this.couchdb_helper.insertNewestSeqNumber(id,seq,status, rev, parent_rev, parent_id);
            }

        }
        catch (Exception e){
            logger.error("ERROR! {}", e.getMessage());
            this.couchdb_helper.insertNewestSeqNumber(id,seq,-1);
        }
    }

    public void delete(DocumentChange change, String id, String rev, int seq){
        try{
            deleteFromElastic(id);
            this.couchdb_helper.insertNewestSeqNumber(id,seq, 1 );
        }
        catch (Exception e){
            this.couchdb_helper.insertNewestSeqNumber(id,seq, -1 );
            logger.error("[delete failed!]");
        }
    }

    public void runIncre(int max_size){
        int seq_num = this.couchdb_helper.getLastSeqNumber(); //181364;
        //seq_num = 181364;
        //boolean ret = this.couchdb_helper.isHaveChangeSinceSeqNum( seq_num );
        //System.out.println("ret = " + ret);
        //String couchdb_name = DbTableNames.KGCouchDB.CONCEPT_TB;
        List<DocumentChange> changes = this.couchdb_helper.getChanges(seq_num, max_size, true);
        for(DocumentChange change : changes) {
            // cur doc
            String id = change.getId();
            String rev = change.getRevision();
            int seq = change.getSequence();
            logger.info("rev  = {}", rev);
            /*
            delete
             */
            boolean isDeleted = change.isDeleted();

            if(isDeleted){
                delete(change, id, rev, seq);
            }
            else {
                insertOrModify(change, id, rev, seq);
            }
        }
    }


    /*************************************
    MAIN ............................................................................
     */
    public static void main(String[] args){
        new ServerConfigurer().initElasticSearch_v2();
        /*
        CouchDbConnector concept_con = new CouchDBKGFactory().getConceptCon();
        CouchdbChangeHelper couchdb_helper = new CouchdbChangeHelper(concept_con, new MySqlDao(),
                DbTableNames.DBType.KG, DbTableNames.TBType.CONCEPT);
                */
        CouchDBDOCFactory couch_factory = new CouchDBDOCFactory();
        CouchDbConnector scene_con = couch_factory.getSceneCon();
        CouchdbChangeHelper scene_helper = new CouchdbChangeHelper(scene_con, new MySqlDao(),
                DbTableNames.DBType.DOC, DbTableNames.TBType.SCENE);
        ElasticChangeHelper es_helper = new ElasticChangeHelper(new ElasticSearchDao(), IndexTypeNames.ESType.SCENE);

        DocUploadHelper docUploadHelper = new DocUploadHelper( new MySqlDao() );

        BaseIncreMode incre_runner = new BaseIncreMode(couch_factory, scene_helper, es_helper, docUploadHelper);
        int idx = 0;
        int MAX_RUN = Integer.MAX_VALUE;
        while (idx < MAX_RUN && incre_runner.hasNext()){
            incre_runner.runIncre( 10 );
            TransactionHelper.GCAndSleep(2000);
            idx += 1;
            logger.info("------------->>>>>>>>>>>>>> run times = {}\n\n", (idx));
        }


    }



}
